use std::collections::HashMap;

use futures::StreamExt;
use wither::{
    bson::{doc, to_document, Document},
    mongodb::options::{FindOneAndUpdateOptions, ReturnDocument},
    Model,
};

use crate::{
    mdb::{MClient, MDB},
    middlewares::session::{CacheData, ExpirableCacheData},
    models::cache_sync::CacheSyncModel,
    AppError, AppResult, ArcConfig, Config,
};
#[allow(unused)]
#[derive(Debug, Clone)]
pub struct SessionService {
    mdb: MDB,
    m_client: MClient,
    config: ArcConfig,
}
/// ## 将hash_map转换为json::Value
/// 会遍历hash_map然后尝试转换value到json::Value，如果失败就直接以字符串的形式加入到返回的map中
fn hash_map_to_json(hash_map: HashMap<String, String>) -> serde_json::Value {
    let mut json_map = serde_json::Map::new();
    for (key, value) in hash_map {
        if let Ok(value) = serde_json::from_str(&value) {
            json_map.insert(key, value);
        } else {
            json_map.insert(key, serde_json::Value::String(value));
        }
    }
    serde_json::Value::Object(json_map)
}
fn document_to_hash_map(doc: Document) -> HashMap<String, String> {
    let mut hash_map = HashMap::new();
    for (key, value) in doc {
        if let Some(value) = value.as_str() {
            hash_map.insert(key.to_string(), value.to_string());
        } else {
            hash_map.insert(
                key.to_string(),
                serde_json::to_string(&value).unwrap_or_default(),
            );
        }
    }
    hash_map
}
impl SessionService {
    pub fn new(config: ArcConfig, mdb: MDB, m_client: MClient) -> Self {
        Self {
            mdb,
            m_client,
            config,
        }
    }
    /// ## 更新或者创建Session
    pub async fn upsert_sessin<T: Into<HashMap<String, String>>>(
        &self,
        id: String,
        expired_time: chrono::DateTime<chrono::Utc>,
        data: T,
    ) -> AppResult<CacheSyncModel> {
        let data: HashMap<String, String> = data.into();
        let value = to_document(&hash_map_to_json(data)).unwrap_or_default();
        let coll = CacheSyncModel::collection(&*self.mdb);
        let options = FindOneAndUpdateOptions::builder()
            .return_document(ReturnDocument::After)
            .upsert(true)
            .build();
        let model = coll
            .find_one_and_update(
                doc! {
                    "c_id":&id
                },
                doc! {
                    "$set":{
                        "value":value,
                        "expired_time":wither::bson::DateTime::from_chrono(expired_time)
                    },
                    "$setOnInsert":{
                        "c_id":id
                    }
                },
                options,
            )
            .await?
            .map(CacheSyncModel::instance_from_document)
            .and_then(Result::ok)
            .ok_or_else(|| AppError::other_error("更新缓存失败"))?;

        Ok(model)
    }
    pub async fn delete_session(&self, id: String) -> AppResult<Option<CacheSyncModel>> {
        Ok(CacheSyncModel::find_one_and_delete(
            &*self.mdb,
            doc! {
                "c_id":id,
            },
            None,
        )
        .await?)
    }
}

pub fn open_session_tree(db: &sled::Db, config: &Config) -> Result<sled::Tree, sled::Error> {
    db.open_tree(&config.cookie.namespace)
}

pub async fn session_gc(db: &sled::Db, session_name: &str) -> AppResult<()> {
    let tree = db.open_tree(session_name)?;
    let now = chrono::Utc::now();
    let expired = tree
        .iter()
        .filter(|v| match v {
            Ok((_, value)) => {
                let cache: CacheData =
                    if let Ok(c) = serde_json::from_slice(value.to_vec().as_slice()) {
                        c
                    } else {
                        // 所有的session都应该是CacheData结构，不一致的直接删除
                        return true;
                    };
                match cache {
                    CacheData::Persistent(_) => false,
                    CacheData::Expirable(e_data) => e_data.expired_time < now,
                }
            }
            Err(_) => false,
        })
        .map(|v| v.unwrap_or_default())
        .collect::<Vec<_>>();
    log::debug!("过期的sessio数为: {}", expired.len());
    for (e_key, _) in expired {
        tree.remove(e_key)?;
    }
    log::debug!("移除过期的session结束");
    Ok(())
}

pub async fn session_sync(mdb: MDB, db: sled::Db, session_name: &str) -> AppResult<()> {
    let tree = db.open_tree(session_name)?;

    let mut local_ids: Vec<String> = Vec::with_capacity(tree.len());

    for item in tree.iter() {
        if let Ok((key, _)) = item {
            if let Ok(key_str) = String::from_utf8(key.to_vec()) {
                local_ids.push(key_str);
            } else {
                continue;
            }
        } else {
            continue;
        }
    }

    let mut cursor = CacheSyncModel::find(&*mdb, None, None).await?;
    let mut delete_ids = local_ids.clone();
    while let Some(Ok(next)) = cursor.next().await {
        if local_ids.contains(&next.c_id) {
            let cache_value = CacheData::Expirable(ExpirableCacheData {
                expired_time: next.expired_time.to_chrono(),
                value: document_to_hash_map(next.value),
            });
            let cache_value_string = serde_json::to_string(&cache_value).unwrap_or_default();
            tree.insert(next.c_id.as_bytes(), cache_value_string.as_bytes())?;
            delete_ids
                .iter()
                .position(|v| v == &next.c_id)
                .map(|v| delete_ids.remove(v));
        }
    }
    for id in delete_ids {
        tree.remove(id.as_bytes())?;
    }
    Ok(())
}
